package com.lmf.cloud.springbootkafka.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

import java.util.HashMap;
import java.util.Map;

/**
 * @Description TODO
 * @Author LiuMingFu
 * @Date 2024/8/11 15:13
 */
//@Configuration
public class KafkaConfig {

    /**
     * 监听器工厂
     *
     * @return
     */
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> concurrencyFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        // 配置额外消费者参数，使用该监听器工厂类的地方，这些参数会覆盖全局配置的相同参数 todo
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(6);
        return factory;
    }

    @Bean
    public ConsumerFactory<? super Integer, ? super String> consumerFactory() {
        Map<String, Object> consumerConfigs = new HashMap<>();
        // 配置额外消费者参数，使用该监听器工厂类的地方，这些参数会覆盖全局配置的相同参数 todo
        return new DefaultKafkaConsumerFactory<>(consumerConfigs);
    }
}
